package com.etlmaster.executor.steps.datax;

import java.util.HashMap;
import java.util.Map;

import com.etlmaster.core.DataSourceDefine;
import com.etlmaster.core.DataSourceHolder;
import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;

/**
 * @author Wugf
 * @date 2016-03-08
 */
@StepInstruction("hivewriter")
public class HiveWriterStep extends Step {
	
	private HiveWriterStepMeta meta;
	public HiveWriterStep(StepMeta meta) {
		super(meta);
		this.meta=(HiveWriterStepMeta)meta;
		// TODO Auto-generated constructor stub
	}

	@Override
	protected int handle() {
		if(meta.getHiveTableName()==null || meta.getHiveTableName().trim()==""){
			return StepState.RUNNING_FAILED;
		}
		
		if(meta.getIsPartTable()=="true"){
			if(meta.getPartWhere().trim().length()<2)
				meta.log("ERROR","分区值为空");
				return StepState.RUNNING_FAILED;
		}
		Map<String, String> data = new HashMap<String, String>();
		DataSourceDefine def = DataSourceHolder.getInstance()
				.getDataSourceDefine(meta.getDsName());

		if(def==null){
			meta.log("ERROR","未分配数据库资源");
			return StepState.RUNNING_FAILED;
		}
		
		data.put("dir", DataCollectionUtils.getProperties("tempDatxHdfsDir"));
		
		data.put("url", def.getUrl());
		data.put("password", def.getPassword());
		data.put("username", def.getUsername());
		
		if(meta.getFieldSplit()==null){
			meta.log("ERROR","请选择分割符号，支持tab,逗号分割符号");
			return StepState.RUNNING_FAILED;
		}
		data.put("field_split", meta.getFieldSplit());
		data.put("prefix_filename",meta.getHiveTableName() );
		data.put("hiveTableName",meta.getHiveTableName() );
		data.put("file_type", meta.getFile_type());
		data.put("encoding", meta.getEncoding());
		data.put("concurrency", meta.getConcurrency());
		data.put("partWhere", meta.getPartWhere());
		Map<String,String> rtnMap= DataxWriter.runDatax(this.hashCode(), meta.getPluginName(), meta.getVariable("readerjobconf").toString(), data);
		meta.getStepState().getReturnValue().put("logInfo", rtnMap.get("jobInfo"));
		return Integer.valueOf(rtnMap.get("status"));
	}

	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
